package jwebtest.rpc.rpc;

import java.util.ArrayList;
import java.util.Collections;
import java.util.Comparator;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicInteger;

import jwebtest.rpc.util.RpcUtil;
import jwebtest.rpc.zk.ZNode;
import jwebtest.rpc.zk.ZkApi;

import org.apache.zookeeper.CreateMode;

import com.alibaba.fastjson.JSON;
import com.jweb.util.FormatUtil;

public abstract class IRpcFramework {
	protected String host;// 客户端访问服务端地址
	protected int port;// 服务端抛露端口
	protected int timeout;// 客户端访问服务端超时时间[单位毫秒]
	protected String defVersion = "0.0.1";
	// == 服务提供者 ============================
	/**
	 * 服务实例[存放接口名与服务对象之间的映射关系]
	 */
	protected Map<String, Object> implServer = new ConcurrentHashMap<String, Object>();

	/**
	 * 服务提供者
	 */
	protected IRpcFramework(int port, int timeout) {
		if (port <= 0 || port > 65535)
			throw new IllegalArgumentException("Invalid port " + port);
		this.port = port;
		this.timeout = timeout;
	}

	/**
	 * 发布服务
	 */
	public abstract boolean addExport(final Object server);

	public abstract boolean addExport(final Object server, int priority, int weight);

	public abstract boolean addExport(final Object server, int priority, int weight, String version);

	// == 服务消费者 ============================

	/**
	 * 服务消费者
	 */
	protected IRpcFramework(String host, int port, int timeout) {
		if (host == null || host.length() == 0)
			throw new IllegalArgumentException("Host == null!");
		if (port <= 0 || port > 65535)
			throw new IllegalArgumentException("Invalid port " + port);
		this.host = host;
		this.port = port;
		this.timeout = timeout;
	}

	/**
	 * 服务消费者
	 */
	protected IRpcFramework(int timeout) {
		this.timeout = timeout;
	}

	/**
	 * 引用服务
	 */
	public abstract <T> T reference(Class<T> _interface);

	public abstract <T> T reference(Class<T> _interface, String version);

	// ==Zookeeper使用==================================================================
	private final String SP = ":";
	public String ZNodePath = "/RPC_Parent";// 默认注册中心父节点
	public String localHost = RpcUtil.getLocalRealIp();// 本地IP
	protected ZkApi zkApi;// 注册中心
	private volatile Map<String, List<ZNode>> serverList = new ConcurrentHashMap<String, List<ZNode>>();// zookeeper所有节点

	public boolean addRegister(ZkApi zkapi) {
		if (zkapi != null)
			this.zkApi = zkapi;
		else
			return false;
		return true;
	}

	/**
	 * 添加注册中心
	 */
	public boolean addRegister(String zkAddress, int timeout) {
		try {
			zkApi = new ZkApi();
			zkApi.createConnection(zkAddress, timeout);
			if (!zkApi.exists(ZNodePath))
				zkApi.createPath(ZNodePath, "--RPC framework for the parent node--", CreateMode.PERSISTENT);
			return true;
		} catch (Exception e) {
			e.printStackTrace();
		}
		return false;
	}

	/**
	 * 创建新的服务节点
	 */
	public void createNode(ZNode zn, int type) {
		String path = "";
		if (type == 1) {
			path = ZNodePath + "/" + zn.getNodeName() + SP + "Provider" + SP + zn.getHost() + SP + zn.getPort();
		} else {
			path = ZNodePath + "/" + zn.getNodeName() + SP + "Consumer" + SP + zn.getHost() + SP + zn.getPort();
		}
		zkApi.createPath(path, FormatUtil.toJSONString(zn), CreateMode.EPHEMERAL);
	}
	/**
	 * 初始化远端服务地址列表【观察者模式】
	 */
	private AtomicInteger threadRun = new AtomicInteger(0);

	public synchronized void initDomainList() {
		try {
			List<String> nodes = zkApi.getChildren(ZNodePath);
			// --获取新列表
			Map<String, List<ZNode>> mapTemp = new HashMap<String, List<ZNode>>();
			for (String node : nodes) {
				if(node.split(SP)[1].equals("Provider")){
					ZNode zn = JSON.parseObject(zkApi.readData(ZNodePath + "/" + node), ZNode.class);
					String key = node.split(SP)[0];
					List<ZNode> ls = mapTemp.get(key);
					if (ls == null)
						ls = new ArrayList<ZNode>();
					ls.add(zn);
					// 排序优先级
					Collections.sort(ls, new Comparator<ZNode>() {
						@Override
						public int compare(ZNode o1, ZNode o2) {
							return o1.getPriority() - o2.getPriority();// 正序，优先级越小越优先
						}
					});
					mapTemp.put(key, ls);
				}
			}
			serverList = mapTemp;// 跟新对象
			System.err.println("node data: " + FormatUtil.toJSONString(serverList));
			this.serverList = serverList;
			// ==开启客户端轮询===
			if (threadRun.get() == 0) {
				threadRun.addAndGet(1);
				new Thread("The client polling to monitor") {
					@Override
					public void run() {
						while (true) {
							try {
								Thread.currentThread().sleep(30 * 1000);
								System.out.println("==重新加载 服务列表");
								initDomainList();
							} catch (Exception e) {
								e.printStackTrace();
							}
						}
					}
				}.start();
			}
		} catch (Exception e) {
			e.printStackTrace();
		}
	}

	/**
	 * 获取远端服务
	 * 
	 * @param interfaceName
	 *            接口名
	 * @param retry
	 *            重试次数[建议3次 ]
	 * @return
	 * @throws Exception
	 */
	public ZNode getDomainServer(String interfaceName, int retry) throws Exception {
		if (retry >= 0) {
			if (zkApi != null) {
				if (serverList.size() == 0) {
					initDomainList();
					return getDomainServer(interfaceName, --retry);
				}
				List<ZNode> ls = serverList.get(interfaceName);
				List<ZNode> temp = new ArrayList<ZNode>();
				temp.add(ls.get(0));
				int weight = ls.get(0).getWeight();
				for (int i = 1; i < ls.size(); i++) {
					if (ls.get(i).getPriority() == ls.get(i - 1).getPriority()) {
						weight += ls.get(i).getWeight();
						temp.add(ls.get(i));
					} else
						break;
				}
				if (temp.size() == 0)
					return ls.get(0);
				else {
					int r = FormatUtil.getRandomInt(weight);
					int ms = 0;
					for (int i = 0; i < temp.size(); i++) {
						ms += temp.get(i).getWeight();
						if (ms >= r) {
							return temp.get(i);
						}
					}
				}
			}
		}
		// --抛出异常，找不到远端服务
		throw new Exception("This service [" + interfaceName + "] can not find, " + "please check whether the remote service providers start."
				+ "From the zookeeper address[" + zkApi.getZK().toString() + "]");
	}

	/**
	 * 关闭注册中心连接[]
	 */
	public void closeRegister() {
		if (zkApi != null)
			zkApi.close();
	}
}
